package com.avicit.zipkinkafkaesserver.service;

import com.avicit.zipkinkafkaesserver.config.ConfigDB;
import com.avicit.zipkinkafkaesserver.config.JobPropertiesConfig;
import com.avicit.zipkinkafkaesserver.dao.ZipkinInterfaceAverageDurationDao;
import com.avicit.zipkinkafkaesserver.dao.es.ZipkinInterfaceAverageDurationEsDao;
import com.avicit.zipkinkafkaesserver.dao.es.entity.ZipkinInterfaceAverageDurationES;
import com.avicit.zipkinkafkaesserver.dao.es.entity.ZipkinInterfaceCountES;
import com.avicit.zipkinkafkaesserver.dto.ZipkinInterfaceAverageDurationDto;
import com.avicit.zipkinkafkaesserver.entity.ZipkinInterfaceAverageDuration;
import com.avicit.zipkinkafkaesserver.entity.ZipkinInterfaceDuration;
import com.avicit.zipkinkafkaesserver.entity.ZipkinUpdateMsg;
import com.avicit.zipkinkafkaesserver.enums.ESIndexsEnum;
import com.avicit.zipkinkafkaesserver.enums.JobTypeEnum;
import com.avicit.zipkinkafkaesserver.enums.ListTypeEnum;
import com.avicit.zipkinkafkaesserver.factory.IDFactory;
import com.avicit.zipkinkafkaesserver.utils.TimeUtils;
import com.avicit.zipkinkafkaesserver.vo.RealtimeAvgDurationChartDataVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import scala.Tuple2;
import zipkin2.Span;

import javax.persistence.criteria.*;
import java.io.IOException;
import java.sql.Time;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;


/**
 * 接口平均耗时统计
 */
@Slf4j
@Service
public class ZipkinInterfaceAverageDurationService {

    @Autowired
    private SpanService spanService;
    @Autowired
    private ZipkinInterfaceAverageDurationEsDao zipkinInterfaceAverageDurationEsDao;
    @Autowired
    private JobPropertiesConfig jobPropertiesConfig;
    @Autowired
    private ZipkinInterfaceAverageDurationDao zipkinInterfaceAverageDurationDao;
    @Autowired
    private ZipkinUpdateMsgService zipkinUpdateMsgService;


    public boolean save(List<Tuple2<String, Double>> tuple2List) {
        List<ZipkinInterfaceAverageDurationES> countVos = tuple2List.stream()
                .map(tuple2 -> {
                    String[] split = tuple2._1.split("@");
                    double avg = tuple2._2;
                    return ZipkinInterfaceAverageDurationES.builder()
                            .interfaceIp(split[0])//取ip
                            .serverName(split[1])//取服务名
                            .interfacePath(split[3])//取接口路径
                            .interfaceAverageDuration((long)avg)//平均耗时
//                            .createTime(System.currentTimeMillis())//取创建时间
                            .queryKey(new StringBuffer(split[0]).append("@")
                                    .append(split[1]).append("@")
                                    .append(split[2]).append("@")
                                    .append(split[3]).append("@")
                                    .toString())
                            .build();
                })
                .collect(Collectors.toList());
        //批量写入es
        String indexPrefix = ESIndexsEnum.ZipkinInterfaceAverageDuration.getIndexPrefix();
        String index = new StringBuffer(indexPrefix).append("-").append(TimeUtils.nowDateString()).toString();
//        boolean save = this.save(index, indexPrefix, countVos);
        return false;
    }


    /**
     * 统计平均耗时
     * @param startTime
     * @param endTime
     * @param min
     * @param max
     * @param topN
     * @return
     */
    public List<ZipkinInterfaceAverageDurationES> query(long startTime, long endTime, long min, long max, int topN) {
        String indexPrefix = ESIndexsEnum.ZipkinInterfaceAverageDuration.getIndexPrefix();
        String index = new StringBuffer(indexPrefix).append("-").append(TimeUtils.nowDateString()).toString();
        return zipkinInterfaceAverageDurationEsDao.query(startTime, endTime, min, max, topN,index, indexPrefix);
    }



    /**
     * 开始时间获取
     * @return
     * @throws IOException
     */
    public Long initStartTime() throws IOException {
        ZipkinInterfaceAverageDurationES durationEsByEndTime = zipkinInterfaceAverageDurationEsDao.findByEndTime(index(), type());
        if (durationEsByEndTime.getEsEndTime() == null) {//当天的没有，查询昨天的
            String yestarday = TimeUtils.backFor(1);
            durationEsByEndTime = zipkinInterfaceAverageDurationEsDao.findByEndTime(index(yestarday),type());
        }

        ConfigDB configDB = jobPropertiesConfig.getConfigDB();

        //是否是系统启动状态
        String startTimeString = configDB.isStatus()?configDB.getStartTime(): new StringBuffer(TimeUtils.nowDateString()).append(" ").append("00:00:00").toString();
        long startTimeFromConfig = TimeUtils.dateTimeToMilliSecond(startTimeString, TimeUtils.DatePattern.ALLTIME);
        long currentTimeMillis = System.currentTimeMillis();
        long startTime = startTimeFromConfig>= currentTimeMillis ? currentTimeMillis :startTimeFromConfig;
        return durationEsByEndTime.getEsEndTime() == null//这里有两种情况是null ，1 是系统刚启动，2是时间到新的一天。如果是新的一天，configDB.isStatus() = false ，开始时间取00:00:00
                ? startTime
                : durationEsByEndTime.getEsEndTime()+1;
    }

    /**
     * index 按天 创建
     * @param date
     * @return
     */
    private String index(String date) {
        final String index = new StringBuffer(type()).append("-").append(date).toString();
        return index;
    }

    /**
     * index 的前缀作为 type
     * @return
     */
    private String type() {
        final String indexPrefix = ESIndexsEnum.ZipkinInterfaceAverageDuration.getIndexPrefix();
        return indexPrefix;
    }

    /**
     * index 按天 创建,默认当前日期
     * @return
     */
    private String index() {
        final String index = new StringBuffer(type()).append("-").append(TimeUtils.nowDateString()).toString();
        return index;
    }

    public void updateStartTime(long endTime) throws IOException {
        ZipkinInterfaceAverageDurationES zipkinInterfaceAverageDurationES = zipkinInterfaceAverageDurationEsDao.findByEndTime(index(),type());
        if (zipkinInterfaceAverageDurationES.getId() == null) {//没有查询，就新增
            log.info("update zipkinInterfaceCountES id is null and save endTime :{}",TimeUtils.timestamToDatetime(endTime));
            log.info("current endTime :{}",endTime);
            zipkinInterfaceAverageDurationES = ZipkinInterfaceAverageDurationES.builder().build();
            zipkinInterfaceAverageDurationES.setEsEndTime(endTime);
            zipkinInterfaceAverageDurationEsDao.save(index(),type(),zipkinInterfaceAverageDurationES);
            return;
        }
        if (endTime>=System.currentTimeMillis()) {log.info("current endTime :{}",endTime);return;}
        Map<String, Object> updateMap = new HashMap<>();
        updateMap.put("esEndTime", endTime);
        log.info("update zipkinInterfaceCountES  endTime :{}",TimeUtils.timestamToDatetime(endTime));
        zipkinInterfaceAverageDurationEsDao.update(updateMap,index(),type(),zipkinInterfaceAverageDurationES.getId());
    }

    /**
     * 保存计算 结果到es
     * @param averageDurationList
     * @param startTime
     * @param endTime
     * @return
     */
    public boolean saveToEs(List<Tuple2<String, Double>> averageDurationList, Long startTime, long endTime, Long jobStartTime , Long jobEndTime) {
        try {
            List<ZipkinInterfaceAverageDurationES> countVos = averageDurationList.stream()
                    .map(tuple2 -> {
                        double interfaceAverageDuration= tuple2._2;
                        String[] split = tuple2._1.split("@");
                        return ZipkinInterfaceAverageDurationES.builder()
                                .interfaceIp(split[0])//取ip
                                .serverName(split[1])//取服务名
                                .interfacePath(split[3])//取接口路径
                                .esEndTime(endTime)
                                .esStartTime(startTime)
                                .queryEndTime(jobEndTime)
                                .queryStartTime(jobStartTime)//记录记录生成的时间段
                                .interfaceAverageDuration((long)interfaceAverageDuration)//平均耗时
                                .queryKey(new StringBuffer(split[0]).append("@")
                                        .append(split[1]).append("@")
                                        .append(split[2]).append("@")
                                        .append(split[3]).append("@")
                                        .toString())
                                .build();
                    })
                    .collect(Collectors.toList());

            //批量写入es
            String type = type();
            String index = index();
            boolean save = this.save(index, type, countVos);
            if (!save) log.error("save averageDurationList to es error :{}", false);
            return save;
        } catch (Exception e) {
            log.error("save averageDurationList to es error :{}", e.getLocalizedMessage());
            return false;
        }
    }

    public boolean save(String index, String type, List<ZipkinInterfaceAverageDurationES> zipkinInterfaceCountList) throws IOException {
        return zipkinInterfaceAverageDurationEsDao.save(index, type, zipkinInterfaceCountList);
    }

    /**
     * 从es 获取一天的平均耗时统计
     * @return
     */
    public List<ZipkinInterfaceAverageDuration> getCountResultFromEs() {
        //es 聚合查询，统计一天的调用总次数，返回，今天统计昨天的
        String indexPrefix = ESIndexsEnum.ZipkinInterfaceAverageDuration.getIndexPrefix();
        LocalDate yesterday = LocalDate.now().plusDays(-1);//昨天的日期
        String index = new StringBuffer(indexPrefix).append("-").append(yesterday).toString();
        List<ZipkinInterfaceAverageDurationDto> countResultFromEs = zipkinInterfaceAverageDurationEsDao.getCountResultFromEs(index, type());
        List<ZipkinInterfaceAverageDuration> collect = countResultFromEs.stream().filter(item->!StringUtils.isEmpty(item.getInterfaceKey())).map(averageDurationDto -> {
            String interfaceKey = averageDurationDto.getInterfaceKey();
            if (!StringUtils.isEmpty(interfaceKey)) {
                String[] split = interfaceKey.split("@");
                ZipkinInterfaceAverageDuration count = ZipkinInterfaceAverageDuration.builder()
                        .serverName(split[0])//服务名
                        .interfaceMethods(split[1])//请求方式
                        .interfacePath(split[2])//请求路径
                        .interfaceAverageDuration(averageDurationDto.getInterfaceAverageDuration())
                        .queryDate(yesterday.toString())//设置该批次的所属日期 ：当前日期的前一天
                        .build();
                return count;
            }
            return null;
        }).collect(Collectors.toList());
        return collect;
    }

    public void saveAllToDb(List<ZipkinInterfaceAverageDuration> countResultFromEs) {
        try {
            List<ZipkinInterfaceAverageDuration> result = new ArrayList<>();
            countResultFromEs.forEach(item->{//有就更新
                ZipkinInterfaceAverageDuration andServerName = zipkinInterfaceAverageDurationDao.findByQueryDateAndInterfacePathAndServerName(item.getQueryDate(), item.getInterfacePath(), item.getServerName());
                if (StringUtils.isEmpty(andServerName)) {
                    item.setId(IDFactory.uuid());
                    result.add(item);
                }else {
                    andServerName.setInterfaceAverageDuration(item.getInterfaceAverageDuration());
                    zipkinInterfaceAverageDurationDao.save(andServerName);
                }
            });
            zipkinInterfaceAverageDurationDao.saveAll(result);
           // this.doSaveUpdateMsg(countResultFromEs);
        }catch (Exception e) {
            log.error("save interface Average Duration error : {}", e.getLocalizedMessage());
           // doErrorSaveUpdateMsg(countResultFromEs);
        }

    }

    /**
     * 接口统计信息批量成功写入时，记录 更新信息
     * @param countResultFromEs
     */
    private void doSaveUpdateMsg(List<ZipkinInterfaceAverageDuration> countResultFromEs) {
        if (countResultFromEs.size() == 0) {return;}
        String queryDate = countResultFromEs.get(0).getQueryDate();
        ZipkinUpdateMsg build = ZipkinUpdateMsg.builder().build();
        build.setJobType(JobTypeEnum.ZipkinInterfaceAverageDuration.getDescription());
        build.setUpdateStatus(true);//该批次的接口统计次数写入成功
        build.setQueryDate(queryDate);
        build.setEsIndex(index(queryDate));
        build.setEsType(type());
        zipkinUpdateMsgService.save(build);
    }

    /**
     * 更新异常记录 信息
     * @param countResultFromEs
     */
    private void doErrorSaveUpdateMsg(List<ZipkinInterfaceAverageDuration> countResultFromEs) {
        String jobType = JobTypeEnum.ZipkinInterfaceAverageDuration.getDescription();
        //属于哪天的数据
        if (countResultFromEs.size() == 0) {return;}
        String queryDate = countResultFromEs.get(0).getQueryDate();
        ZipkinUpdateMsg zipkinUpdateMsg = zipkinUpdateMsgService.findByQueryDateAndUpdateStatus(queryDate, jobType,false);
        if (StringUtils.isEmpty(zipkinUpdateMsg)) {//当前日期没有异常的更新
            ZipkinUpdateMsg build = ZipkinUpdateMsg.builder().build();
            build.setJobType(jobType);
            build.setUpdateStatus(false);//该批次的接口统计次数写入失败
            build.setQueryDate(queryDate);
            build.setEsIndex(index(queryDate));
            build.setErrorCount(1);
            build.setEsType(type());
            zipkinUpdateMsgService.save(build);
            return;
        }
        //错误次数加一
        zipkinUpdateMsg.setErrorCount(zipkinUpdateMsg.getErrorCount()+1);
        zipkinUpdateMsgService.save(zipkinUpdateMsg);
    }

    public List<ZipkinInterfaceAverageDuration> interfaceAvgDuration(String queryDate) {
        return zipkinInterfaceAverageDurationDao.findAll(new Specification<ZipkinInterfaceAverageDuration>() {
            @Override
            public Predicate toPredicate(Root<ZipkinInterfaceAverageDuration> root, CriteriaQuery<?> query, CriteriaBuilder criteriaBuilder) {
                Predicate rootPredicate = null;
                if (!StringUtils.isEmpty(queryDate)) {
                    rootPredicate = criteriaBuilder.equal(root.get("queryDate"), queryDate);
                }
                return rootPredicate;
            }
        });
    }

    public List<ZipkinInterfaceAverageDuration> interfaceAvgDuration(String queryDateFrom, String queryDateTo, String serverName, String interfacePath) {
        return zipkinInterfaceAverageDurationDao.findAll((Specification<ZipkinInterfaceAverageDuration>) (root, criteriaQuery, criteriaBuilder) -> {
            Predicate predicateRoot = criteriaBuilder.greaterThanOrEqualTo(root.get("queryDate"),queryDateFrom);//大于等于
            Predicate lessThanOrEqualTo = criteriaBuilder.lessThanOrEqualTo(root.get("queryDate"), queryDateTo);//小于等于
            predicateRoot = criteriaBuilder.and(predicateRoot,lessThanOrEqualTo);//取区间
            if (!StringUtils.isEmpty(serverName)) {//服务名
                Predicate serverNamePredicate = criteriaBuilder.equal(root.get("serverName").as(String.class), serverName);
                predicateRoot = criteriaBuilder.and(predicateRoot,serverNamePredicate);
            }
            if (!StringUtils.isEmpty(interfacePath)) {//接口
                predicateRoot = criteriaBuilder.and(predicateRoot,criteriaBuilder.equal(root.get("interfacePath").as(String.class),interfacePath));
            }
            //根据接口耗时倒排
            Order desc = criteriaBuilder.desc(root.get("interfaceAverageDuration"));
            criteriaQuery.where(predicateRoot).orderBy(desc);
            return null;
        },PageRequest.of(0,100)).getContent();
    }

    public List<RealtimeAvgDurationChartDataVo> getRealtimeAvgDurationChartData(String queryDateFrom, String queryDateTo, String serverName, String interfacePath) throws IOException {
        queryDateFrom = StringUtils.isEmpty(queryDateFrom)?TimeUtils.backFor(1): queryDateFrom;
        queryDateTo = StringUtils.isEmpty(queryDateTo)?TimeUtils.nowDateString():queryDateTo;

        long startTime = TimeUtils.dateTimeToMilliSecond(queryDateFrom + " 00:00:00", TimeUtils.DatePattern.ALLTIME);

        long endTime  = TimeUtils.dateTimeToMilliSecond(queryDateTo + " 23:59:59", TimeUtils.DatePattern.ALLTIME);
        String type = type();//index 前缀
        String index = new StringBuffer(type).append("-").append("*").toString();
        List<ZipkinInterfaceAverageDurationES> zipkinInterfaceAverageDurationES = zipkinInterfaceAverageDurationEsDao.getRealtimeAvgDurationChartData(startTime, endTime, serverName, interfacePath, index, type);
        List<RealtimeAvgDurationChartDataVo> collect = zipkinInterfaceAverageDurationES.stream().map(item ->
                RealtimeAvgDurationChartDataVo.builder()
                        .time(item.getQueryStartTime())
                        .reqs(item.getInterfaceAverageDuration())
                        .serverName(item.getServerName())
                        .interfaceAverageDuration(item.getInterfaceAverageDuration())
                        .interfaceName(item.getInterfacePath())
                        .queryStartTime(item.getQueryStartTime())
                        .queryEndTime(item.getQueryEndTime())
                        .build()
                        .setQueryEndTimeString()
                        .setQueryStartTimeString()).collect(Collectors.toList());

        return collect;
    }
}
